package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.BooleanUtil;
import com.hmdp.dto.Result;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.RedisIdWorker;
import com.hmdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    // 自动注入秒杀优惠券服务接口，用于处理优惠券相关的业务逻辑
    @Autowired
    private ISeckillVoucherService seckillVoucherService;

    // 自动注入Redis ID生成器，用于生成全局唯一的订单ID
    @Autowired
    private RedisIdWorker redisIdWorker;

    // 自动注入StringRedisTemplate，用于操作Redis中的字符串类型数据，如执行lua脚本
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    // 自动注入RedissonClient，用于分布式锁等高级功能，确保秒杀过程中的数据一致性
    @Autowired
    private RedissonClient redissonClient;

    // 定义Redis脚本，用于执行秒杀逻辑，如检查秒杀资格和扣减库存
    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;

    static {
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua")); // 设置lua脚本的位置
        SECKILL_SCRIPT.setResultType(Long.class); // 设置脚本返回结果的类型为Long
    }

    // 定义一个阻塞队列，用于存储待处理的订单信息
//    private BlockingQueue<VoucherOrder> orderQueue = new ArrayBlockingQueue<>(1024 * 1024);

    // 定义一个线程池，用于异步处理订单
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

    // 初始化方法，在Spring容器加载完成后执行，启动订单处理线程
    @PostConstruct
    private void init() {
        SECKILL_ORDER_EXECUTOR.execute(new VoucherOrderHandler());
    }

    // 定义一个内部类，实现Runnable接口，用于处理订单
    private class VoucherOrderHandler implements Runnable {
        // 队列名称
        String queueName = "stream.orders";

        @Override
        public void run() {
            while (true) {
                try {
                    // 从消息队列中获取订单信息
                    // 使用XREADGROUP命令，指定消费者组、消费者名称、读取参数（读取数量、是否阻塞、阻塞时间）以及队列名称和结束标识（最近一条未消费的消息）
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("group1", "c1"), // 指定消费者组和消费者名称
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)), // 读取参数：读取一个消息，阻塞时间为2秒
                            StreamOffset.create(queueName, ReadOffset.lastConsumed()) // 指定队列名称和结束标识
                    );

                    // 判断消息是否获取成功
                    if (list == null || list.isEmpty()) {
                        // 获取失败，说明没有消息，进行下一次循环
                        continue;
                    }

                    // 获取成功，执行下单
                    // 从list中取出订单
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> value = record.getValue(); // 获取消息的内容，这是一个键值对形式的Map
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true); // 将Map转换为VoucherOrder对象

                    // 处理订单，包括创建订单和更新数据库等
                    handleVoucherOrder(voucherOrder);

                    // ACK确认，通知Redis这个消息已经被处理完毕
                    stringRedisTemplate.opsForStream().acknowledge(queueName, "group1", record.getId());
                } catch (Exception e) {
                    // 记录处理订单时的异常信息
                    log.error("处理订单异常:", e);
                    handlePendingList();
                }
            }
        }

        private void handlePendingList() {
            while (true) {
                try {
                    // 从Pending List中获取订单信息
                    // 使用XREADGROUP命令，指定消费者组、消费者名称、读取参数（读取数量）以及队列名称和结束标识（从队列开头读取）
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("group1", "c1"), // 指定消费者组和消费者名称
                            StreamReadOptions.empty().count(1), // 读取参数：读取一个消息，不阻塞
                            StreamOffset.create(queueName, ReadOffset.from("0")) // 指定队列名称和结束标识，从队列开头读取
                    );

                    // 判断消息是否获取成功
                    if (list == null || list.isEmpty()) {
                        // 获取失败，说明没有消息，跳出循环
                        break;
                    }

                    // 获取成功，执行下单
                    // 从list中取出订单
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> value = record.getValue(); // 获取消息的内容，这是一个键值对形式的Map
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true); // 将Map转换为VoucherOrder对象

                    // 处理订单，包括创建订单和更新数据库等
                    handleVoucherOrder(voucherOrder);

                    // ACK确认，通知Redis这个消息已经被处理完毕
                    stringRedisTemplate.opsForStream().acknowledge(queueName, "group1", record.getId());
                } catch (Exception e) {
                    // 记录处理pending-list订单时的异常信息
                    log.error("处理pending-list订单异常:", e);
                    try {
                        // 暂停20毫秒，防止过于频繁地处理异常消息
                        Thread.sleep(20);
                    } catch (InterruptedException ex) {
                        // 如果线程被中断，抛出运行时异常
                        throw new RuntimeException(ex);
                    }
                }
            }
        }
    }
    /*
    private class VoucherOrderHandler implements Runnable {
        @Override
        public void run() {
            while (true) {
                try {
                    // 从队列中获取订单信息
                    VoucherOrder voucherOrder = orderQueue.take();
                    // 处理订单，包括创建订单和更新数据库等
                    handleVoucherOrder(voucherOrder);
                } catch (Exception e) {
                    // 记录处理订单时的异常信息
                    log.error("处理订单异常:", e);
                }
            }
        }
    }

     */

    // 代理对象，用于在事务方法内部调用其他事务方法时保持事务的一致性
    private IVoucherOrderService proxy;

    // 处理订单的方法，包括获取分布式锁、创建订单和释放锁等步骤
    private void handleVoucherOrder(VoucherOrder voucherOrder) {
        // 使用Redisson客户端获取分布式锁，确保同一用户不能重复下单
        RLock lock = redissonClient.getLock("lock:order:" + voucherOrder.getUserId());
        // 尝试获取锁，如果获取失败则直接返回，避免重复下单
        boolean isLock = lock.tryLock();
        if (!isLock) {
            log.error("不允许重复下单");
            return;
        }
        try {
            // 通过代理对象调用创建订单的方法，确保事务的一致性
            proxy.creatVoucherOrder(voucherOrder);
        } finally {
            // 释放锁，确保锁资源被正确释放，避免死锁问题
            lock.unlock();
        }
    }


    /**
     * 秒杀优惠券的方法
     *
     * @param voucherId 优惠券ID，用于指定要秒杀的优惠券
     * @return 秒杀结果，包括成功或失败的信息以及订单ID（如果成功）
     */
    @Override
    @Transactional
    public Result seckillVoucher(Long voucherId) {
        Long userId = UserHolder.getUser().getId(); // 获取当前用户的ID
        long orderId = redisIdWorker.globalIDMint("order"); // 生成全局唯一的订单ID
        // 执行lua脚本，检查秒杀资格和扣减库存
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.emptyList(), // 脚本没有用到key参数，所以传递空列表
                voucherId.toString(), userId.toString(), String.valueOf(orderId) // 传递优惠券ID、用户ID、订单id作为脚本的参数
        );
        // 根据脚本执行结果判断秒杀是否成功
        if (result != 0) {
            // 结果不为0，表示秒杀失败，根据结果返回相应的失败信息
            return Result.fail(result == 1 ? "库存不足" : "不能重复下单");
        }

        // 获取当前类的代理对象，用于在事务方法内部调用其他事务方法
        proxy = (IVoucherOrderService) AopContext.currentProxy();
        // 返回订单ID，表示秒杀成功
        return Result.ok(orderId);
    }
    /*
    @Override
    @Transactional
    public Result seckillVoucher(Long voucherId) {
        Long userId = UserHolder.getUser().getId(); // 获取当前用户的ID
        // 执行lua脚本，检查秒杀资格和扣减库存
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.emptyList(), // 脚本没有用到key参数，所以传递空列表
                voucherId.toString(), userId.toString() // 传递优惠券ID和用户ID作为脚本的参数
        );
        // 根据脚本执行结果判断秒杀是否成功
        if (result != 0) {
            // 结果不为0，表示秒杀失败，根据结果返回相应的失败信息
            return Result.fail(result == 1 ? "库存不足" : "不能重复下单");
        }
        // 结果为0，表示秒杀成功，生成订单ID并保存订单信息到队列中
        long orderId = redisIdWorker.globalIDMint("order"); // 生成全局唯一的订单ID
        VoucherOrder voucherOrder = new VoucherOrder()
                .setId(orderId)
                .setUserId(userId)
                .setVoucherId(voucherId);
        orderQueue.add(voucherOrder); // 将订单信息添加到队列中等待处理
        // 获取当前类的代理对象，用于在事务方法内部调用其他事务方法
        proxy = (IVoucherOrderService) AopContext.currentProxy();
        // 返回订单ID，表示秒杀成功
        return Result.ok(orderId);
    }
     */

    /*
    @Override
    @Transactional // 开启事务，确保方法中的数据库操作要么全部成功，要么全部回滚
    public Result seckillVoucher(Long voucherId) {
        // 查询优惠券信息
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        // 判断秒杀是否尚未开始
        if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
            return Result.fail("秒杀尚未开始");
        }
        // 判断秒杀是否已经结束
        if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
            return Result.fail("秒杀已经结束");
        }
        // 判断库存是否充足
        if (voucher.getStock() < 1) {
            return Result.fail("库存不足");
        }
        Long userId = UserHolder.getUser().getId(); // 获取当前用户的ID
        // 对用户ID进行字符串锁定，防止同一用户多次下单
        // 创建锁
        // 注释掉的原有synchronized代码，替换为Redisson分布式锁
        // synchronized (userId.toString().intern()) {
        // SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
        RLock lock = redissonClient.getLock("lock:order:" + userId); // 使用Redisson客户端获取分布式锁
        // 尝试获取锁
        boolean isLock = lock.tryLock();
        if (!isLock) {
            // 获取锁失败，返回错误信息
            return Result.fail("不允许重复下单");
        }
        try {
            // 获取代理对象，用于事务控制
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            // 创建优惠券订单
            return proxy.creatVoucherOrder(voucherId);
        } finally {
            // 释放锁，确保锁资源被释放，避免死锁
            lock.unlock();
        }
    }
    */

    /**
     * 创建优惠券订单
     *
     * @param voucherOrder 优惠券信息
     */
    @Override
    @Transactional
    public void creatVoucherOrder(VoucherOrder voucherOrder) {
        // 确保一人一单
        Long userId = voucherOrder.getUserId();  // 获取当前用户的ID
        // 查询用户是否已经下过该优惠券的订单
        Long count = query().eq("user_id", userId).eq("voucher_id", voucherOrder.getVoucherId()).count();
        if (count > 0) {
            log.info("用户已购买过一次！");
            return;
        }
        // 调用秒杀优惠券服务接口的方法，尝试扣减库存
        boolean success = seckillVoucherService.update()
                .setSql("stock = stock - 1") // 设置更新语句，将库存数量减一
                .eq("voucher_id", voucherOrder.getVoucherId()) // 添加条件，只更新指定优惠券ID的记录
                .gt("stock", 0) // 添加条件，只有当库存数量大于0时才进行更新
                .update(); // 执行更新操作，返回布尔值表示更新是否成功
        if (!success) {
            log.info("库存不足，扣减失败");
            return;
        }
        // 库存扣减成功，创建订单并保存到数据库中
        save(voucherOrder);
    }
}